package org.teemor.demo.kafka.springclient.config;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.ProducerListener;

import java.util.HashMap;
import java.util.Map;

/**
 * @Description:
 * @author zhoulk
 * @date: 2020/2/13
 */
@Configuration
public class KafkaProduceConfig {

    @Bean
    KafkaTemplate<Integer, String> kafkaTemplate() {
        KafkaTemplate<Integer,String> kafkaTemplate = new KafkaTemplate<>(producerFactory());
        kafkaTemplate.setDefaultTopic("test");//调用sendDefault()方法时，默认发送到该主题
        kafkaTemplate.setProducerListener(producerListener());
        return kafkaTemplate;
    }

    @Bean
    public ProducerFactory<Integer, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public ProducerListener<Integer, String> producerListener () {
        ProducerListener<Integer, String> producerListener = new ProducerListener<Integer, String>() {
            @Override
            public void onSuccess(ProducerRecord producerRecord, RecordMetadata recordMetadata) {
                System.out.println("消息发送成功 + 1");
            }

            @Override
            public void onSuccess(String topic, Integer partition, Integer key, String value, RecordMetadata recordMetadata) {
                System.out.println("消息发送成功 + 2");
            }

            @Override
            public void onError(ProducerRecord producerRecord, Exception exception) {
                System.out.println("消息发送失败 + 1");
            }

            @Override
            public void onError(String topic, Integer partition, Integer key, String value, Exception exception) {
                System.out.println("消息发送失败 + 2");
            }
        };
        return producerListener;
    }

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "39.108.107.152:9092,39.108.107.152:9091,39.108.107.152:9093");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.ACKS_CONFIG, "0");
        // See https://kafka.apache.org/documentation/#producerconfigs for more properties
        return props;
    }
}
